package com.liyueheng.app.dataWarehouseDetail.behavior

import com.liyueheng.util.{ConfigLoader, SaveAsTable, SparkConf}
import org.apache.spark.sql.functions._

object BounceAndConversionRate {
  def analyze(): Unit = {
    println("------------------ 分析跳失率和转化率 -----------------")
    val spark = SparkConf.createSparkSession("BounceConversion")
    val dwd = ConfigLoader.getString("databases.dwd")
    val dws = ConfigLoader.getString("databases.dws")

    import spark.implicits._

    val df = spark.table(s"$dwd.user_act")

    // 各用户行为统计
    val userStats = df.groupBy("user")
      .agg(
        sum(when($"act_type" === 2, 1).otherwise(0)).alias("buy_num"),
        sum(when($"act_type" === 5, 1).otherwise(0)).alias("cart_num"),
        sum(when($"act_type" === 3, 1).otherwise(0)).alias("fav_num"),
        sum(when($"act_type" === 4, 1).otherwise(0)).alias("comment_num")
      )

    // 跳失用户定义：四类行为都是 0
    val bounceUsers = userStats.filter($"buy_num" === 0 && $"cart_num" === 0 && $"fav_num" === 0 && $"comment_num" === 0)
      .count()

    val totalUsers = userStats.count()

    val bounceRate = (bounceUsers.toDouble / totalUsers) * 100
    val conversionRate = 100.0 - bounceRate

    val result = Seq(
      ("跳失率", f"$bounceRate%.2f%%"),
      ("转化率", f"$conversionRate%.2f%%")
    ).toDF("name", "value")

    SaveAsTable.saveAsTable(result, s"$dws.detail_bounce_conversion_rate")
    SparkConf.stopSparkSession(spark)
  }
}
